use std::collections::{VecDeque};
use std::io::Cursor;
use std::sync::Arc;
use std::time::Duration;
use async_trait::async_trait;
use bytes::{Buf, Bytes, BytesMut};
use log::{error, info};
use tokio::io::{split, AsyncReadExt, AsyncWriteExt, ReadHalf, WriteHalf};
use tokio::net::TcpStream;
use tokio::sync::{broadcast, mpsc};
use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
use tokio::time::{interval};
use crate::{send, spawn, ChannelId, RpcContext, RpcEvent, RpcResult};
use crate::connection_manager::event::{ConnectionManagerDataEvent, IdleListener, ReadData};

pub struct RpcConnection {
    id: ChannelId,
    buff_size: usize,
    context: RpcContext,
    event_receiver: UnboundedReceiver<Box<dyn RpcEvent>>,
    idle_listeners: Arc<Vec<Box<dyn IdleListener>>>,
}

#[async_trait]
pub trait ConnectionHandle: Send + Sync {
    async fn connect_open(&mut self, id: ChannelId, sender: UnboundedSender<Bytes>)-> RpcResult<()>;
    
    fn connect_clone(&mut self, id: ChannelId);

    fn handle(&mut self, id: ChannelId, rpc_context: &mut RpcContext,  data: Vec<u8>) -> RpcResult<()>;
}

impl RpcConnection {

    pub fn new(id: ChannelId,
               buff_size: usize,
               context: RpcContext,
               event_receiver: UnboundedReceiver<Box<dyn RpcEvent>>,
               idle_listeners: Arc<Vec<Box<dyn IdleListener>>>
               ) -> RpcConnection {
        RpcConnection {
            id,
            buff_size,
            context,
            event_receiver,
            idle_listeners,
        }
    }

    pub async fn run<F: FnOnce()>(&mut self, stream: TcpStream, mut handler: impl
        ConnectionHandle, open_end_fn: F, mut shutdown: broadcast::Receiver<()>) -> RpcResult<()> {

        let mut context = self.context.clone();
        let (stream_reader, stream_writer) = split(stream);
        let (read_bytes_sender, mut read_bytes_receiver) = mpsc::unbounded_channel::<Bytes>();
        let (write_bytes_sender, write_bytes_receiver) = mpsc::unbounded_channel();

        let shutdown_clone = shutdown.resubscribe();
        spawn(connection_reader(self.id, stream_reader, read_bytes_sender, self.buff_size, context.clone(), shutdown_clone));

        let shutdown_clone = shutdown.resubscribe();
        spawn(connection_writer(self.id, stream_writer, write_bytes_receiver, shutdown_clone));

        handler.connect_open(self.id, write_bytes_sender).await?;

        open_end_fn();

        let mut interval = interval(Duration::from_secs(60));

        loop {
            tokio::select! {
                data = read_bytes_receiver.recv() =>{
                    match data {
                        None => {break;}
                        Some(data) => {
                            info!("receive data, length: {}", data.len());
                            let do_handle_result = handler.handle(self.id, &mut context, data.to_vec());
                            if let Err(e) = do_handle_result {
                                error!("handler error: {}", e);
                            }
                        }
                    }
                }
                event = self.event_receiver.recv() =>{
                    match event {
                        None => {
                            break;
                        }
                        Some(event) => {
                            let result = event.consume(self.context.clone());
                            if let Err(e) = result {
                                error!("event handle error: {}", e);
                            }
                        }
                    }
                }
                _ = interval.tick() => {  // 检查连接哪些是idle
                    println!("idle listener length2: {}", self.idle_listeners.len());
                    if !self.idle_listeners.is_empty() {
                        let idle_listeners = self.idle_listeners.clone();
                        let idles = self.idle_listeners.iter().map(|l| l.to_idle_data()).collect::<Vec<_>>();
                        let event = ConnectionManagerDataEvent::Idles {data: idles, sender: self.context.sender(),listeners: idle_listeners};
                        let result = self.context.send_data(Box::new(event));
                        if let Err(e) = result {
                            error!("interval send error: {}", e);
                        }
                        println!("interval send success!!!!!!!!!!!!")
                    }
                }
                _ = shutdown.recv() => {
                    println!("Connection {:?}: shutdown", self.id);
                    break;
                }
            }
        }

        println!("Connection {:?}: disconnected", self.id);
        handler.connect_clone(self.id);
        Ok(())
    }
}

async fn connection_reader(
    channel_id: ChannelId,
    mut stream: ReadHalf<TcpStream>,
    read_bytes_sender: UnboundedSender<Bytes>,
    buff_size: usize,
    mut context: RpcContext,
    mut shutdown: broadcast::Receiver<()>,
) -> RpcResult<()> {
    let mut buffer = BytesMut::with_capacity(buff_size);

    loop {
        tokio::select! {
            bytes_read = stream.read_buf(&mut buffer) => {
                match bytes_read {
                    Ok(bytes_read) => {
                        println!("connectId {} : {} bytes read, buffer size: {}", channel_id.id,
                            bytes_read, buffer.capacity());
                        if bytes_read == 0 { // 流断开
                            break;
                        }

                        let read_data = ReadData::new(channel_id, bytes_read);
                        context.send_data(Box::new(ConnectionManagerDataEvent::Read(read_data)))?;
                        let packet = parse(&mut buffer);
                        if let Some(packet) = packet {
                            if !send(&read_bytes_sender, packet) {
                                error!("send receive data error");
                                break;
                            }
                            info!("receive packet and send success");
                        }
                    }
                    Err(error) => {
                        error!("connection reader error: {}", error);
                        break
                    }
                }
            }
            _ = shutdown.recv() => {
                break;
            }
        }
    }

    info!("Connection {:?}: Reader disconnected", channel_id);
    Ok(())
}

async fn connection_writer(
    connection_id: ChannelId,
    mut stream: WriteHalf<TcpStream>,
    mut packets_to_send: UnboundedReceiver<Bytes>,
    mut shutdown: broadcast::Receiver<()>,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let mut send_queue = VecDeque::new();

    loop {
        tokio::select! {
            packet = packets_to_send.recv() =>{
                match packet {
                    None => {break}
                    Some(packet) => {
                        send_queue.push_back(packet);
                        while let Ok(packet) = packets_to_send.try_recv() {
                            send_queue.push_back(packet);
                        }

                        for packet in send_queue.drain(..) {
                            stream.write_all(&packet[..]).await?;
                            info!("connection write data length: {}", packet.len());
                        }
                    }
                }
            }
            _ = shutdown.recv() => {
                break;
            }
        }
    }

    info!("Connection {:?}: Writer disconnected", connection_id);
    Ok(())
}

fn parse(bytes_mut: &mut BytesMut) -> Option<Bytes> {
    let mut cursor = Cursor::new(&bytes_mut[..]);
    if cursor.remaining() < 4 {
        return None;
    }
    let len = cursor.get_u32();
    if cursor.remaining() < len as usize {
        return None;
    }

    let mut bytes = bytes_mut.split_to((len + 4) as usize);
    let data = bytes.split_off(4);
    let bytes = data.freeze();
    info!("data len: {}: {}, bytes_mut len: {}, bytes_mut capacity: {}", bytes.len(), len, bytes_mut.len(), bytes_mut.capacity());
    Some(bytes)
}

